package com.shiguiwu.springmybatis.javabase.thread;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @description: 手写并发工具
 * @author: stone
 * @date: Created by 2021/6/24 9:42
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springmybatis.thread
 */
public class TaskDisposeUtils {

    //并发数
    public static final int POOL_SIZE;
    static {
        POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);
    }

    /**
     * 并发处理
     * @param taskList 任务列表
     * @param consumer 消费者
     * @param <T> 类型
     * @throws InterruptedException 异常
     */
    public static <T> void dispose( List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        dispose(true, POOL_SIZE, taskList, consumer);
    }

    /**
     * 并发处理
     * @param isMoreThread 是否并发
     * @param poolSize 线程数
     * @param taskList 任务列表
     * @param consumer 消费者
     * @param <T> 类型
     * @throws InterruptedException 异常
     */
    public static <T> void dispose(boolean isMoreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        if (isMoreThread && poolSize > 1) {
            poolSize = Math.min(poolSize, taskList.size());
            ExecutorService service = null;
            try {
                service = Executors.newFixedThreadPool(poolSize);

                ExecutorService finalService = service;
                CountDownLatch countDownLatch = new CountDownLatch(taskList.size());
                taskList.forEach(t -> finalService.execute(() -> {
                    try {
                        consumer.accept(t);
                    } finally {
                        countDownLatch.countDown();
                    }

                }));
                countDownLatch.await();
            }  finally {
                if (service != null) {
                    service.shutdown();
                }
            }
        }
        else {
            taskList.forEach(consumer::accept);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
        dispose(list, l-> {
            try {
                TimeUnit.SECONDS.sleep(l);
                System.out.println("结束=======================>");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

}
